package com.mls.guardian.master;

import com.mls.guardian.container.MasterContainer;
import com.mls.guardian.protocol.RemoteBody;
import com.mls.guardian.protocol.RemoteCommand;
import com.mls.guardian.protocol.body.AppRestartBody;
import com.mls.guardian.remoting.client.ClientConnect;
import com.mls.guardian.remoting.netty.RemoteTools;
import io.netty.channel.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;

/**
 * Created by jl on 15/11/3.
 */
public class ClientDispatch {
    private static final Logger log = LoggerFactory.getLogger(ClientDispatch.class);
    private final ConcurrentHashMap<String, ClientConnect> clientTables = new ConcurrentHashMap<String, ClientConnect>();
    private MasterContainer masterContainer;

    public ClientDispatch(MasterContainer masterContainer) {
        this.masterContainer = masterContainer;
    }

    public void addClient(String client, ClientConnect connect) {
        clientTables.put(client, connect);
    }

    public Object removeClient(String client, ClientConnect connect) {
        Iterator<String> it = clientTables.keySet().iterator();

        while (it.hasNext()) {
            String key = it.next();

            if (key.equals(client) && clientTables.get(key).equals(connect)) {
                return clientTables.remove(key);
            }
        }

        return null;
    }

    public Map getClient() {
        return clientTables;
    }


    public List<RemoteBody> sendRequestToClients(RemoteCommand request, List<String> addrs, long timeoutMills) throws Exception {
        List<RemoteBody> responseList = new ArrayList<RemoteBody>();
        //全部发送消息
        if (addrs == null) {
            Iterator<String> it = clientTables.keySet().iterator();
            while (it.hasNext()) {
                String addr = it.next();
                RemoteBody remoteBody = sendRequestToClientSync(request, addr, timeoutMills);
                responseList.add(remoteBody);
            }
        } else {
            for (String addr : addrs) {
                Set<String> clientSet = clientTables.keySet();
                if (clientSet.contains(addr)) {
                    RemoteBody remoteBody = sendRequestToClientSync(request, addr, timeoutMills);
                    responseList.add(remoteBody);
                }
            }
        }

        return responseList;
    }

    public List<Future<RemoteBody>> sendRequestToClientsByThreadPools(RemoteCommand request, List<String> addrs, long timeoutMills, ExecutorService executorService) throws Exception {
        List<Future<RemoteBody>> futures = new ArrayList<Future<RemoteBody>>();

        if (addrs == null) {
            Iterator<String> it = clientTables.keySet().iterator();
            while (it.hasNext()) {
                String addr = it.next();
                futures.add(executorService.submit(new Callable<RemoteBody>() {
                    @Override
                    public RemoteBody call() throws Exception {
                        RemoteBody responseBody = (AppRestartBody) sendRequestToClientSync(request, addr, timeoutMills);
                        log.info("id[{}] code[{}] slave-{} 处理结束 返回", addr, request.getRequestId(), request.getCode());
                        return responseBody;
                    }
                }));
            }

        } else {
            for (String addr : addrs) {
                Set<String> clientSet = clientTables.keySet();
                if (clientSet.contains(addr)) {
                    futures.add(executorService.submit(new Callable<RemoteBody>() {
                        @Override
                        public RemoteBody call() throws Exception {
                            RemoteBody responseBody = sendRequestToClientSync(request, addr, timeoutMills);
                            log.info("id[{}] code[{}] slave-{} 处理结束 返回", addr, request.getRequestId(), request.getCode());
                            return responseBody;
                        }
                    }));
                }
            }
        }

        return futures;
    }


    private RemoteBody sendRequestToClientSync(RemoteCommand request, String addr, long timeoutMills) throws Exception {
        RemoteCommand requestNew = RemoteTools.getRequestRemoteCmd(request.getCode(), request.getRemoteBody());

        final Channel channel = (io.netty.channel.Channel) clientTables.get(addr).getConnect();
        log.info("id[{}] code[{}] 发送给客户端{} channel-{}", requestNew.getRequestId(), request.getCode(), addr, channel);
        RemoteCommand response = masterContainer.getServer().sendMessageSync(channel
                , requestNew, timeoutMills);
        return response.getRemoteBody();
    }

    /**
     * 只发送不管结果
     *
     * @param request
     * @param addrs
     * @return
     * @throws Exception
     */
    public void sendRequestToClientOneWay(RemoteCommand request, List<String> addrs) throws Exception {
        List<RemoteBody> responseList = new ArrayList<RemoteBody>();
        //全部发送消息
        if (addrs == null) {
            Iterator<String> it = clientTables.keySet().iterator();
            while (it.hasNext()) {
                String addr = it.next();
                sendRequestToClientOneWay(request, addr);
            }
        } else {
            for (String addr : addrs) {
                Set<String> clientSet = clientTables.keySet();
                if (clientSet.contains(addr)) {
                    sendRequestToClientOneWay(request, addr);
                }
            }
        }
    }


    /**
     * 只发送不管结果
     *
     * @param request
     * @param addr
     * @throws Exception
     */
    public void sendRequestToClientOneWay(RemoteCommand request, String addr) throws Exception {
        RemoteCommand requestNew = RemoteTools.getRequestRemoteCmd(request.getCode(), request.getRemoteBody());

        final Channel channel = (io.netty.channel.Channel) clientTables.get(addr).getConnect();
        log.info("id[{}] code[{}] 发送给客户端{} channel-{}", requestNew.getRequestId(), request.getCode(), addr, channel);
        masterContainer.getServer().sendMessageOneWay(channel, requestNew);
    }

//    public RemoteBody sendRequestToClients(RemoteCommand request, String addr) throws Exception {
//        Iterator<String> it = clientTables.keySet().iterator();
//        RemoteBody remoteBody = null;
//
//        RemoteCommand requestNew = RemoteTools.getRequestRemoteCmd(request.getCode(), request.getRemoteBody());
//        log.info("restart app 发送给客户端{} 请求id是{}", addr, requestNew.getRequestId());
//        RemoteCommand response = masterContainer.getServer().sendMessageSync(
//                (io.netty.channel.Channel) clientTables.get(addr).getConnect(), requestNew, -1);
//
//        remoteBody = response.getRemoteBody();
//
//        return remoteBody;
//    }
}
